package com.mashibing.rocketmq;

/**
 * @Author: JLL
 * @Date: 2020/3/2 17:48
 */
import java.util.List;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;

public class Consumer2 {

    public static void main(String[] args)throws Exception {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("xoxogp");

        consumer.setNamesrvAddr("192.168.150.133:9876");

        // 每个consumer 关注一个topic

        // topic 关注的消息的地址
        // 过滤器 * 表示不过滤
        // tag selector 在一个group中的消费者，都不能随便变，要保持统一
        consumer.subscribe("myTopic003", "TAG-B");

        consumer.registerMessageListener(new MessageListenerConcurrently() {

            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {

                for (MessageExt msg : msgs) {

                    System.out.println(new String(msg.getBody()));;
                }
                // 默认情况下 这条消息只会被 一个consumer 消费到 点对点
                // message 状态修改
                // ack
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        ////这是默认的，集群模式。
//        当使用集群消费模式时，MQ 认为任意一条消息只需要被集群内的任意一个消费者处理即可。
        consumer.setMessageModel(MessageModel.CLUSTERING);
        consumer.start();

        System.out.println("Consumer TAG-B start...");







    }

}
